from vo import TriggerDAGItem, ActivateDAGItem
import requests
import custom_config
from random_id import generate_random_id
from response import create_response

async def trigger_dag_impl(item: TriggerDAGItem):
    url = custom_config.AIRFLOW_API_URL + f"/api/v1/dags/{item.dag_id}/dagRuns"
    data = {
        "dag_run_id": generate_random_id()
    }
    headers = {
        'Authorization': 'Basic YWRtaW46YWRtaW4=',
        'Content-Type': 'application/json'
    }
    response = requests.post(url, json=data, headers=headers)
    if response is not None:
        if response.status_code == 200:
            data = response.json()  # 假设响应内容是JSON格式
            return create_response(
                success=True,
                message="trigger success",
                data={"res": data}
            )
        else:
            return create_response(
                success=False,
                message='fail, code：' + str(response.status_code),
            )
    else:
        return create_response(
            success=False,
            message='fail',
        )

async def activate_dag_impl(item: ActivateDAGItem):
    url = custom_config.AIRFLOW_API_URL + f"/api/v1/dags/{item.dag_id}"
    data = {
        "is_paused": False
    }
    headers = {
        'Authorization': 'Basic YWRtaW46YWRtaW4=',
        'Content-Type': 'application/json'
    }
    response = requests.patch(url, json=data, headers=headers)
    if response is not None:
        if response.status_code == 200:
            data = response.json()  # 假设响应内容是JSON格式
            return create_response(
                success=True,
                message="activate success",
                data={"res": data}
            )
        else:
            return create_response(
                success=False,
                message='fail, code：' + str(response.status_code),
            )
    else:
        return create_response(
            success=False,
            message='fail',
        )